package com.huitone.smspfm.interfacz.mo.deliver;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;

import com.bin.common.utils.DateTimeUtils;
import com.bin.common.utils.StringUtils;
import com.huitone.smspfm.interfacz.logger.ExtLogger;
import com.huitone.smspfm.interfacz.mo.SvrNodeClientsManager;
import com.huitone.smspfm.interfacz.core.AppStaticConfigs;
import com.huitone.smspfm.interfacz.dao.entity.SvrNodeDef;
import com.huitone.smspfm.interfacz.model.MoDealHash;
import com.huitone.smspfm.interfacz.model.SmsDeliver;
import com.huitone.smspfm.interfacz.utils.SMSContentUtils;
import com.huitone.smspfm.redis.model.AppCanstant;
import com.huitone.smspfm.redis.pipeline.PipelineOpts.ABExcuteCommands;
import com.huitone.smspfm.socket.client.SmsClient;
import com.huitone.smspfm.socket.protocol.sgip.constant.CommandID;
import com.huitone.smspfm.socket.protocol.sgip.constant.Constant;
import com.huitone.smspfm.socket.protocol.sgip.message.Message;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Deliver;
import com.huitone.smspfm.socket.protocol.sgip.message.msghead.Head;

/**
 * @Title DeliverManager.java
 * @Package com.huitone.smspfm.interfacz.core
 * @Description TODO(用一句话描述该文件做什么)
 * @author ybliang
 * @date 2017年4月4日 上午11:53:23
 * @version V1.0
 */
public class SmsDeliverTask extends Thread {
	
	private static final ExtLogger logger = ExtLogger.create(SmsDeliverTask.class);
	
	private ExecutorService executorService = new ThreadPoolExecutor(AppStaticConfigs.smspfmMoPoolSize, 
			AppStaticConfigs.smspfmMoPoolSize, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(AppStaticConfigs.smspfmMoPoolSize * 2), new RejectedExecutionHandler() {
		
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			boolean offer = false;
			while(!executor.isShutdown() && !offer) {
				offer = executor.getQueue().offer(r);
				if (!offer) {
					try {
						TimeUnit.MILLISECONDS.sleep(10);
					} catch (InterruptedException e) {}
				}
			}
		}
	});
	
	private boolean toSleep = false;
	
	private long sleepTime = 1000;
	
	private static final String CurrMoSpeedS = "CurrMoSpeedS";
	private static final String CurrMoSpeedM = "CurrMoSpeedM";
	
	private static final byte[] CURR_MO_SPEEDS_B = AppStaticConfigs.stringRedisSerializer.serialize(CurrMoSpeedS);
	
	private static final byte[] CURR_MO_SPEEDM_B = AppStaticConfigs.stringRedisSerializer.serialize(CurrMoSpeedM);
	
	private byte[] currMoSpeedS_List = AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.RunningParamsHash + ":" + CurrMoSpeedS);
	private byte[] currMoSpeedM_List = AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.RunningParamsHash + ":" + CurrMoSpeedM);
	
	private long currSpeedS = 0;
	
	private long currSpeedM = 0;
	
	private final Object slock = new Object();
	
	private final Object mlock = new Object();
	
	private ArrayBlockingQueue<Map<byte[],byte[]>> speeds = new ArrayBlockingQueue<Map<byte[],byte[]>>(1000);
	
	private boolean keepGoing = true;
	
	@Override
	public void run() {
		
		if (AppStaticConfigs.speedStatisticsOn) {
			
			startUpdateSpeedTasks();
			
			startSpeedWriteTasks();
			
		}
		
		@SuppressWarnings("rawtypes")
		RedisScript<List> script = new DefaultRedisScript<List>(AppStaticConfigs.luaGetMo, List.class);
		
		List<String> keys = new ArrayList<>();
		keys.add(AppStaticConfigs.iSMMoDealList);
		keys.add(AppStaticConfigs.pSMCacheHash);
		
		@SuppressWarnings("rawtypes")
		RedisScript<List> script2 = new DefaultRedisScript<>(AppStaticConfigs.luaMoSendSucc, List.class);
		
		List<String> keys2 = new ArrayList<>();
		keys2.add(AppStaticConfigs.iSMMoDealSerialNoSet);
		keys2.add(AppStaticConfigs.iSMMoRespSerialNoZSet);
		
		while (!this.isInterrupted() && isKeepGoing()) {
			
			executorService.execute(new Runnable() {
				
				@Override
				public void run() {
					
					List<List<String>> lists = AppStaticConfigs.scriptOpts3.getHashList(script, keys, AppStaticConfigs.moReadPerTime + "");
					
					List<SmsDeliver> entities = new ArrayList<SmsDeliver>();
					for (int i = 0 ; i < lists.size() ; i ++) {
						List<String> list = lists.get(i);
						SmsDeliver entity = SmsDeliver.parse(list);
						entities.add(entity);
					}
					
					if (entities.isEmpty()) {
						toSleep = true;
						return;
					}
					
					toSleep = false;
					
					List<MoDealHash> mdhs = new ArrayList<MoDealHash>();
					
					AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {
						
						@Override
						public void excute(RedisConnection connection) {
					
							for (SmsDeliver sd:entities) {
								
								String cacheKey = sd.getSerialno() + ":" + sd.getPhoneNumber();
								
								if (sd.isWellInformed()) {
									
									SvrNodeDef svrNodeDef = AppStaticConfigs.svrNodeDefMap.get(sd.getSrvNode());
									
									if (svrNodeDef != null) {
										
										if (StringUtils.equals(svrNodeDef.getAcceptMo(), "0") || svrNodeDef.getAcceptMo() == null) { // 是否向业务端推送上行短信, 0 推送 1 不推送 2 强制不推送(由于多次连接不成功导致)
											
											SmsClient smsClient = SvrNodeClientsManager.getInstance().getSvrNodeClient(sd.getSrvNode());
											
											if (smsClient != null) {
												
												String orgSrvNode = sd.getOrgSrvNode();
												
												String srvNode = StringUtils.isEmpty(orgSrvNode) ? sd.getSrvNode() : orgSrvNode;
												
												String sendPort = StringUtils.isEmpty(orgSrvNode) ? sd.getSendPort() : "95598" + sd.getSendPort();
													
												Deliver deliver = Deliver.build(srvNode, sendPort, sd.getPhoneNumber(), sd.getRecvTime(), Constant.MessageCodingE.GBK.getValue(), SMSContentUtils.cut70(sd.getSmcontent()));
												
												Head head = Head.build(deliver.length(), CommandID.SGIP_DELIVER, sd.getSerialPart1(), sd.getSerialPart2(), sd.getSerialPart3());
												
												//String serialNo = head.getSerialNo();
												
												/// ---
												
												MoDealHash dealHash = new MoDealHash();
												
												dealHash.setiSMCacheKey(cacheKey);
												dealHash.setiRptSnPart1(head.getSerialNoPart1());
												dealHash.setiRptSnPart2(head.getSerialNoPart2());
												dealHash.setiRptSnPart3(head.getSerialNoPart3());
												dealHash.setiSendTimes(1);
												dealHash.setiLastSendTime(new Date());
												
												mdhs.add(dealHash);
												
												//String hkey = AppStaticConfigs.pMMoDealTmpDlvHash + ":" + serialNo;
												
												//byte[] hkeyb = AppStaticConfigs.stringRedisSerializer.serialize(hkey);
												
												//connection.hMSet(hkeyb, dealHash.getMapB());
												//connection.expire(hkeyb, 3600);
												//connection.zAdd(AppStaticConfigs.iSMMoDealSerialNoSetB, 
												//		System.currentTimeMillis() + AppStaticConfigs.moResendInterval, AppStaticConfigs.stringRedisSerializer.serialize(head.getSerialNo()));
												
												/// ---
												
												Map<byte[], byte[]> map = new HashMap<byte[], byte[]>();
												map.put(AppCanstant.SmsEntity.MoTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(null)));
												
												connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey), map);
												
												Message message = new Message(head, deliver);
												
												logger.delivrd(message.toString());
												
												boolean sendDeliver =smsClient.sendMsg(message);
												
												if (!sendDeliver) {
													logger.dfailed(sd.toString());
												}
													
											} else {
												
												String desc = "没有匹配到业务端:" + sd.getSrvNode();
												
												connection.lPush(AppStaticConfigs.iSMMoEndListB, AppStaticConfigs.stringRedisSerializer.serialize(cacheKey));
												
												Map<byte[], byte[]> map = new HashMap<byte[], byte[]>();
												map.put(AppCanstant.SmsEntity.MoResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
												map.put(AppCanstant.SmsEntity.MoResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
												map.put(AppCanstant.SmsEntity.MoTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(null)));
												
												connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey),map);
											
												logger.dfailed(desc + sd.toString());
											}
										} else {
											if (StringUtils.equals(svrNodeDef.getAcceptMo(), "2")) {
												logger.dfailed("业务端:" + svrNodeDef.getSvrNode() + "当前ACCEPT_MO状态为2,连接不可用,直接写上行短信完成队列:" + sd.toString());
											}
											// 直接写上行完成队列
											AppStaticConfigs.listOpts3.in(AppStaticConfigs.iSMMoEndList, cacheKey);
										}
									} else {
										String desc = "没有匹配到业务端:" + sd.getSrvNode();
										
										connection.lPush(AppStaticConfigs.iSMMoEndListB, AppStaticConfigs.stringRedisSerializer.serialize(cacheKey));
										
										Map<byte[], byte[]> map = new HashMap<byte[], byte[]>();
										map.put(AppCanstant.SmsEntity.MoResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
										map.put(AppCanstant.SmsEntity.MoResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
										map.put(AppCanstant.SmsEntity.MoTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(null)));
										
										connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey),map);
									
										logger.dfailed(desc + sd.toString());
									}
									
								} else {
									String desc = "读取的上行短信信息不全:";
									connection.lPush(AppStaticConfigs.iSMMoEndListB, AppStaticConfigs.stringRedisSerializer.serialize(cacheKey));
									
									Map<byte[], byte[]> map = new HashMap<byte[], byte[]>();
									map.put(AppCanstant.SmsEntity.MoResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
									map.put(AppCanstant.SmsEntity.MoResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
									map.put(AppCanstant.SmsEntity.MoTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(null)));
									
									connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey),map);
								
									logger.dfailed(desc + sd.toString());
								}
							}
							
							if (AppStaticConfigs.speedStatisticsOn) {
								int size = entities.size();
								increaseCurrSpeedS(size);
								increaseCurrSpeedM(size);
							}
						}
					});
					
					List<String> serialnoList = new ArrayList<>();
					
					// 保存重发上行短信需要到的信息
					AppStaticConfigs.pipelineOpts2.save(new ABExcuteCommands() {

						@Override
						public void excute(RedisConnection connection) {
							
							for (MoDealHash mdh:mdhs) {
								
								String hkey = AppStaticConfigs.pMMoDealTmpDlvHash + ":" + mdh.getSerialNo();
								
								byte[] hkeyb = AppStaticConfigs.stringRedisSerializer.serialize(hkey);
								
								connection.hMSet(hkeyb, mdh.getMapB());
								connection.expire(hkeyb, 86400);
								//connection.zAdd(AppStaticConfigs.iSMMoDealSerialNoSetB, 
								//		System.currentTimeMillis() + AppStaticConfigs.moResendInterval, AppStaticConfigs.stringRedisSerializer.serialize(mdh.getSerialNo()));
								serialnoList.add(mdh.getSerialNo() + ":" + (System.currentTimeMillis() + AppStaticConfigs.moResendInterval));
							}
						}
					});
					
					// 保存上行短信序列号的同时判断是否收到响应消息 
					List<List<String>> nfhstrlist = AppStaticConfigs.scriptOpts2.getHashList(script2, keys2, serialnoList.toArray());
					
					List<MoDealHash> nfhentities = new ArrayList<MoDealHash>();
					for (int i = 0 ; i < nfhstrlist.size() ; i ++) {
						List<String> tmplist = nfhstrlist.get(i);
						MoDealHash entity = MoDealHash.parseStr(tmplist);
						nfhentities.add(entity);
					}
					
					if (!nfhentities.isEmpty()) {
						
						AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {

							@Override
							public void excute(RedisConnection connection) {
								
								List<byte[]> moendList = new ArrayList<>();
								
								for (MoDealHash mdh:nfhentities) {
									
									Map<byte[], byte[]> map = new HashMap<>();
									
									map.put(AppCanstant.SmsEntity.MoResult, AppStaticConfigs.stringRedisSerializer.serialize("0"));
									map.put(AppCanstant.SmsEntity.MoResultDesc, AppStaticConfigs.stringRedisSerializer.serialize("success"));
									
									connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + mdh.getiSMCacheKey()), map);
									
									moendList.add(AppStaticConfigs.stringRedisSerializer.serialize(mdh.getiSMCacheKey()));
								}
								
								if (!moendList.isEmpty()) {
									byte[][] bs = moendList.toArray(new byte[1][]);
									connection.lPush(AppStaticConfigs.iSMMoEndListB, bs);
								}
							}
						});
					}
				}
			});
			
			try {
				if (toSleep) {
					Thread.sleep(sleepTime);
				}
			} catch (InterruptedException e) {
				break;
			}
		}
		
		if (executorService != null && !executorService.isShutdown()) {
			executorService.shutdown();
		}
		
		logger.info("上行短信任务退出!");
	}
	
	private void startUpdateSpeedTasks() {
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				while((executorService != null && !executorService.isShutdown())) {
					
					long speed = getAndResetCurrSpeedS();
					
					speeds.offer(Collections.singletonMap(CURR_MO_SPEEDS_B, AppStaticConfigs.stringRedisSerializer.serialize(speed + ":" + DateTimeUtils.dateToStr2(null))));
					
					logger.debug("上行短信转发速度(条/s):" + speed);
					
					try {
						TimeUnit.MILLISECONDS.sleep(1000);
					} catch (InterruptedException e) {}
				}
			}
		}).start();
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				int count = 0;
				
				while((executorService != null && !executorService.isShutdown())) {
					
					if (count >= 60) {
						
						long speed = getAndResetCurrSpeedM();
						
						speeds.offer(Collections.singletonMap(CURR_MO_SPEEDM_B, AppStaticConfigs.stringRedisSerializer.serialize(speed + ":" + DateTimeUtils.dateToStr2(null))));
						
						logger.debug("上行短信转发速度(条/m):" + speed);
						
						count = 0;
						
					} else {
						count ++;
					}
					
					try {
						TimeUnit.MILLISECONDS.sleep(1000);
					} catch (InterruptedException e) {}
				}
			}
		}).start();
	}
	
	private void startSpeedWriteTasks() {
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				while ((executorService != null && !executorService.isShutdown())) {
					
					Map<byte[],byte[]> map = speeds.poll();
					if (map != null && !map.isEmpty()) {
						AppStaticConfigs.pipelineOpts.save(new ABExcuteCommands() {

							@Override
							public void excute(RedisConnection connection) {
								connection.hMSet(AppStaticConfigs.RunningParamsHashB, map);
								// 每秒的速率队列
								byte[] bytes = map.get(CURR_MO_SPEEDS_B);
								if (bytes != null) {
									connection.lPush(currMoSpeedS_List, bytes); 
									connection.lTrim(currMoSpeedS_List, 0, AppStaticConfigs.speedSLength - 1);
								}
								// 每分的速率队列
								byte[] bytem = map.get(CURR_MO_SPEEDM_B);
								if (bytem != null) {
									connection.lPush(currMoSpeedM_List, bytem); 
									connection.lTrim(currMoSpeedM_List, 0, AppStaticConfigs.speedMLength - 1);
								}
							}
						});
					} else {
						try {
							TimeUnit.MILLISECONDS.sleep(1000);
						} catch (InterruptedException e) {}
					}
				}
			}
		}).start();
	}

	@Override
	public void destroy() {
		this.setKeepGoing(false);
		this.interrupt();
	}

	public boolean isKeepGoing() {
		return keepGoing;
	}

	public void setKeepGoing(boolean keepGoing) {
		this.keepGoing = keepGoing;
	}
	
	private long getAndResetCurrSpeedS() {
		long result = 0;
		synchronized (slock) {
			result = currSpeedS;
			currSpeedS = 0;
		}
		return result;
	}

	private void increaseCurrSpeedS(long count) {
		synchronized (slock) {
			this.currSpeedS += count;
		}
	}

	public long getCurrSpeedM() {
		return currSpeedM;
	}
	
	private long getAndResetCurrSpeedM() {
		long result = 0;
		synchronized (mlock) {
			result = currSpeedM;
			currSpeedM = 0;
		}
		return result;
	}

	private void increaseCurrSpeedM(long count) {
		synchronized (mlock) {
			this.currSpeedM += count;
		}
	}
}
